
package com.huaweicloud.reqmt.source;

import com.huaweicloud.reqmt.domain.dws_haydn_self_verification_dura_reqmt_d;

import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Properties;

/**
 * 连接DWS数据库 建立与 dws_haydn_self_verification_dura_reqmt_d 表的连接
 *
 * @author l30028736
 * @since 2022-10-29
 */
public class DwsSourceDws001 extends RichParallelSourceFunction<dws_haydn_self_verification_dura_reqmt_d> {
    private static final Logger logger = LoggerFactory.getLogger(DwsSourceDwd001.class);

    private static PreparedStatement ps;

    private static Connection conn;

    /**
     * 类在初始化的时候创建数据库连接
     */
    public DwsSourceDws001(String username, String password, String url) {
        createConnection(username, password, url);
    }

    /**
     * 建立与DWS的连接
     */
    private void createConnection(String username, String password, String url) {
        Properties props = new Properties();
        props.setProperty("user", username);
        props.setProperty("password", password);
        props.setProperty("ssl", "false");
        try {
            Class.forName("org.postgresql.Driver");
            conn = DriverManager.getConnection(url, props);
            logger.info("Build connection success.");
            String sql =
                "SELECT ds, workspace_id, workspace_name, solu_uuid, solu_name, reqmt_uuid, reqmt_name, case_type, workflow_status, start_time, end_time, dura_trans, dura_seconds FROM dws.dws_haydn_self_verification_dura_reqmt_d";
            ps = conn.prepareStatement(sql);
            logger.info("prepareStatement create success.");
        } catch (ClassNotFoundException | SQLException e) {
            logger.error("Build connection failed");
        }
    }

    /**
     * 执行SQL并发回去
     *
     * @param ctx context
     * @throws SQLException SQL执行异常
     */
    @Override
    public void run(SourceContext<dws_haydn_self_verification_dura_reqmt_d> ctx) throws SQLException {
        ResultSet query = ps.executeQuery();
        while (query.next()) {
            dws_haydn_self_verification_dura_reqmt_d data =
                new dws_haydn_self_verification_dura_reqmt_d(query.getString("ds"), query.getString("workspace_id"),
                    query.getString("workspace_name"), query.getString("solu_uuid"), query.getString("solu_name"),
                    query.getString("reqmt_uuid"), query.getString("reqmt_name"), query.getString("case_type"),
                    query.getString("workflow_status"), query.getString("start_time"), query.getString("end_time"),
                    query.getString("dura_trans"), query.getString("dura_seconds"));
            // 发送数据
            ctx.collect(data);
        }
    }

    /**
     * 程序关闭时会释放链接
     */
    @Override
    public void cancel() {
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                logger.error("Close connect failed");
            }
        }
        if (ps != null) {
            try {
                ps.close();
            } catch (SQLException e) {
                logger.error("Close prepareStatement failed");
            }
        }
    }
}